package personal.wang.handler;

import com.google.protobuf.InvalidProtocolBufferException;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import personal.wang.domain.HeartbeatBean;
import personal.wang.abs.AbstractMessageHandler;
import personal.wang.manager.WorkerManager;
import personal.wang.message.Message;
import personal.wang.worker.HeartbeatThread;

/**
 * @className: BeartbeatHandler
 * @Description:
 * @Author: wangyifei
 * @Date: 2025/2/14 16:53
 */
public class HeartbeatHandler extends AbstractMessageHandler {
    private static Logger logger = LoggerFactory.getLogger(HeartbeatHandler.class);
    private WorkerManager workerManager ;
    private HeartbeatThread run ;
    private int ttl = 4 ;
    public void start(){
        run = new HeartbeatThread(()->{
            // 检查那些终端超时了
            workerManager.getWorkers().forEach((k,v)->{
                if(System.currentTimeMillis() - v.getTimeToLive() > ttl*1000 ){
                    if(!v.getChannel().isOpen()){
                        logger.info("{} endpoint closed" , k);
                    }
                    logger.info("{} endpoint timeout" , k);
                }
            });
        } , ttl);
    }
    {
        start();
    }
    public HeartbeatHandler(){}
    public HeartbeatHandler(WorkerManager workerManager){
       this.workerManager = workerManager ;
    }

    public void setWorkerManager(WorkerManager workerManager) {
        this.workerManager = workerManager;
    }

    public WorkerManager getWorkerManager() {
        return workerManager;
    }

    @Override
    public void handler(Message message , ChannelHandlerContext ctx) {
        HeartbeatBean.Heartbeat deserialize = null;
        try {
            Channel channel = ctx.channel();
            deserialize = HeartbeatBean.Heartbeat.parseFrom(message.getContent());
            workerManager.getWorkers().put(deserialize.getEndpoint() , new WorkerManager.WorkerInfor(System.currentTimeMillis() , ctx.channel()));
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        }
        logger.debug("receive heartbeat , cnt : {} ,  from {} , {} endpoint "
                    , deserialize.getReport().getMessageCnt()
                    , deserialize.getEndpoint().getHost()
                ,deserialize.getEndpoint().getIp()
        );
    }
}
